package com.yugao.fintech.draper.message.queue.client;

import com.yugao.fintech.draper.message.queue.interceptor.RedisMessageInterceptor;
import com.yugao.fintech.draper.message.queue.message.AbstractBroadcastMessage;
import com.yugao.fintech.draper.message.queue.message.AbstractGroupMessage;
import com.yugao.fintech.draper.core.util.JacksonUtil;
import lombok.AllArgsConstructor;
import lombok.Getter;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.RedisTemplate;

/**
 * redis消息队列操作类
 */
@Getter
@AllArgsConstructor
public class RedisMQTemplate extends MessageQueueTemplate<RedisMessageInterceptor> {

    private final RedisTemplate<String, ?> redisTemplate;

    @Override
    protected void doSend(AbstractGroupMessage message) {
        ObjectRecord<String, String> record = StreamRecords.newRecord()
                .ofObject(JacksonUtil.toJsonString(message))
                .withStreamKey(message.getGroup());
        // 发送分组消息
        redisTemplate.opsForStream().add(record);
    }

    @Override
    protected void doSend(AbstractBroadcastMessage message) {
        // 发送广播消息
        redisTemplate.convertAndSend(message.getTopic(), JacksonUtil.toJsonString(message));
    }
}
